package one.bugu.zookeeper.rpc.framework.client.socket;

import one.bugu.zookeeper.rpc.framework.entity.Message;
import com.google.gson.Gson;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created with IntelliJ IDEA.
 * User: LangK
 * Created Date 2019/2/11
 * Time: 17:25
 * Description:
 */
public class ClientSocket extends Thread {

    private Gson gson = new Gson();
    private Logger logger = LoggerFactory.getLogger(ClientSocket.class);

    private ChannelFuture futrue;

    private String serverName;
    private String host;
    private int port;
    private CallBack callBack;

    public ClientSocket(String serverName, String host, int port, CallBack callBack) {
        this.serverName = serverName;
        this.host = host;
        this.port = port;
        this.callBack = callBack;
    }

    /**
     * 打开通道
     *
     * @param host
     * @param port
     */
    public void connection(String host, int port) {
        //worker负责读写数据
        synchronized (ClientSocket.class) {
            EventLoopGroup worker = new NioEventLoopGroup();
            try {
                //辅助启动类
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(worker)
                        .channel(NioSocketChannel.class).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                        .handler(new ChannelInitializer<Channel>() {
                            @Override
                            protected void initChannel(Channel ch) {
                                ch.pipeline().addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                                ch.pipeline().addLast("decoder", new StringDecoder());
                                ch.pipeline().addLast("encoder", new StringEncoder());
                                ch.pipeline().addLast("handler", new ClientHandler());
                            }
                        });
                futrue = bootstrap.connect(host, port).sync();
                //等待客户端链路关闭
                futrue.channel().closeFuture().sync();
            } catch (Exception e) {
                this.callBack.connectionFail(serverName);
                logger.error("RPC长链接客户端异常", e);
            } finally {
                //优雅的退出，释放NIO线程组
                worker.shutdownGracefully().syncUninterruptibly();
            }
        }
    }


    /**
     * 长连接是否可用
     *
     * @return
     */
    public boolean isActive() {
        if (futrue == null || futrue.channel() == null) {
            return false;
        }
        return futrue.channel().isActive();
    }

    /**
     * 发送消息
     *
     * @param message
     */
    public boolean send(Message message) {
        try {
            if (isActive()) {
                futrue.channel().writeAndFlush(message.toString() + "\r\n");
                return true;
            } else {
                logger.error("RPC长链接客户端发送消息失败，连接已断开");
            }
        } catch (Exception e) {
            logger.error("RPC长链接客户端发送消息异常", e);
        }
        return false;
    }

    /**
     * 关闭通道
     */
    public void close() {
        futrue.channel().close();
    }

    @Override
    public void run() {
        connection(host, port);
    }


    class ClientHandler extends SimpleChannelInboundHandler<String> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
            try {
                synchronized (msg) {
                    Message message = gson.fromJson(msg, Message.class);
                    callBack.receiveMessage(message);
                }
            } catch (Exception e) {
                logger.info("服务端返回异常消息:{}", msg);
                logger.error("处理服务端返回的消息异常", e);
                ReferenceCountUtil.release(msg);
            }
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            super.channelActive(ctx);
            logger.info("Client active");
            ClientSocket.this.callBack.connectionSuccess(serverName, ClientSocket.this);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
            logger.info("Client close");
            ClientSocket.this.callBack.connectionFail(serverName);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
            logger.info("Client connection exception");
        }

    }

    /**
     * Socket回调接口
     */
    public static interface CallBack {
        /**
         * 异步返回
         * 连接成功
         */
        void connectionSuccess(String serverName, ClientSocket clientSocket);

        /**
         * 异步返回
         * 连接失败
         */
        void connectionFail(String serverName);

        /**
         * 异步返回
         * 接收到消息
         *
         * @param message
         */
        void receiveMessage(Message message);
    }

}
